package com.gss.plugin.kafka.config;

import com.fasterxml.jackson.databind.deser.std.StringDeserializer;
import org.apache.kafka.clients.CommonClientConfigs;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.common.config.SaslConfigs;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.annotation.EnableKafka;
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
import org.springframework.kafka.config.KafkaListenerContainerFactory;
import org.springframework.kafka.core.ConsumerFactory;
import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
import org.springframework.kafka.listener.ConcurrentMessageListenerContainer;
import org.springframework.kafka.listener.ContainerProperties;

import java.util.HashMap;
import java.util.Map;

/**
 * @author: DYang
 * @since: 2023/8/24
 * @description: 消费者配置
 */
@ConditionalOnProperty(name = "stars.kafka.consumer.enable",havingValue = "true")
@Configuration
@EnableKafka
public class KafkaConsumerConfig {

    @Autowired
    private KafkaProperties kafkaProperties;

    public Map<String, Object> consumerConfigs() {
        KafkaProperties.ConsumerConfig consumerConfig = kafkaProperties.getConsumer();
        Map<String, Object> propsMap = new HashMap<>();
        propsMap.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaProperties.getServers());
        propsMap.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, consumerConfig.isAutoCommit());
        propsMap.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, consumerConfig.getCommitInterval());
        propsMap.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, consumerConfig.getSessionTimeout());
        propsMap.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        propsMap.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        propsMap.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, consumerConfig.getOffsetReset());

        propsMap.put(ConsumerConfig.REQUEST_TIMEOUT_MS_CONFIG,consumerConfig.getRequestTimeout());
        propsMap.put(ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG,consumerConfig.getMaxPollInterval());
        propsMap.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG,consumerConfig.getMaxPollRecords());
        propsMap.put(ConsumerConfig.FETCH_MAX_BYTES_CONFIG,consumerConfig.getFetchMaxBytes());
        propsMap.put(ConsumerConfig.HEARTBEAT_INTERVAL_MS_CONFIG,consumerConfig.getHeartbeatInterval());
        propsMap.put(ConsumerConfig.PARTITION_ASSIGNMENT_STRATEGY_CONFIG,consumerConfig.getPartitionAssignmentStrategy());

        propsMap.put(ConsumerConfig.GROUP_ID_CONFIG, consumerConfig.getGroupId());

        //安全认证
        if (kafkaProperties.isSecurityEnable()) {
            //配置security.protocol
            propsMap.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, kafkaProperties.getSecurityProtocol());
            //配置sasl.mechanism
            propsMap.put(SaslConfigs.SASL_MECHANISM, kafkaProperties.getSaslMechanism());
            //配置sasl.jaas.config
            propsMap.put(SaslConfigs.SASL_JAAS_CONFIG, kafkaProperties.getSaslJaasConfig());
        }
        return propsMap;
    }

    @Bean("kafkaListenerContainerFactory")
    public KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<String, String>> kafkaListenerContainerFactory() {
        ConcurrentKafkaListenerContainerFactory<String, String> factory = new ConcurrentKafkaListenerContainerFactory<>();
        Map<String, Object> configs = consumerConfigs();

        DefaultKafkaConsumerFactory<Object, Object> consumerFactory = new DefaultKafkaConsumerFactory<>(configs);
        factory.setConsumerFactory(consumerFactory);
        factory.setConcurrency(kafkaProperties.getConsumer().getConcurrency());
        factory.getContainerProperties().setPollTimeout(kafkaProperties.getConsumer().getRequestTimeout());
        factory.setBatchListener(true);

        // 手动提交偏移量
        factory.getContainerProperties().setAckMode(ContainerProperties.AckMode.MANUAL);
        return factory;
    }

    @Bean
    public ConsumerFactory<Object, Object> consumerFactory() {
        // 配置ConsumerFactory
        return new DefaultKafkaConsumerFactory<>(consumerConfigs());
    }

}
